package yuekao9.machine;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalBinaryClassBatchOp;
import com.alibaba.alink.operator.batch.sink.CsvSinkBatchOp;
import com.alibaba.alink.operator.batch.utils.UDFBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.batch.sql.SelectBatchOp;
import com.alibaba.alink.operator.common.evaluation.BinaryClassMetrics;
import com.alibaba.alink.operator.common.evaluation.TuningMultiClassMetric;
import com.alibaba.alink.pipeline.PipelineModel;
import com.alibaba.alink.pipeline.classification.DecisionTreeClassifier;
import com.alibaba.alink.pipeline.classification.LogisticRegression;
import com.alibaba.alink.pipeline.tuning.GridSearchCV;
import com.alibaba.alink.pipeline.tuning.MultiClassClassificationTuningEvaluator;
import com.alibaba.alink.pipeline.tuning.ParamGrid;
import org.apache.flink.table.functions.ScalarFunction;

public class Leukaemia {
    public static void main(String[] args) throws Exception {
        //1.导入白血病数据集并进行数据预处理，包括特征缩放和将特征和目标变量分成训练和测试数据集。(（5分）
        String filePath = "data/yk9/白血病数据集.txt";
        String schema
                //Obs    age      sex       bmi      map      tc       ldl      hdl      tch      ltg      glu    y
                //  1  0.03808  0.050680  0.06170  0.02187 -0.04422 -0.03482 -0.04340 -0.00259  0.01991 -0.01765 151
                = "Obs int, age double, sex double, bmi double, map double, tc double, ldl double, hdl double, tch double, ltg double, glu double, y int";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",")
                .setIgnoreFirstLine(true)
                .setSkipBlankLine(true)
                .setLenient(true);
//        csvSource.print();

        SelectBatchOp selectBatchOp = new SelectBatchOp()
                .setClause("age,sex, bmi,map,tc, ldl,hdl,tch,ltg,glu,y")
                .linkFrom(csvSource);
//        selectBatchOp.print();

        UDFBatchOp udfBatchOp = new UDFBatchOp()
                .setFunc(new SubstringFunction())
                .setSelectedCols("y")
                .setOutputCol("y1")
                .linkFrom(selectBatchOp);



        String[] features=new String[]{"age","sex", "bmi","map","tc", "ldl","hdl","tch","ltg","glu"};
        String label="y1";


        //拆分数据集
        SplitBatchOp split = new SplitBatchOp().setFraction(0.8);
        SplitBatchOp trainData = split.linkFrom(udfBatchOp);
        BatchOperator<?> testData = split.getSideOutput(0);
//        System.out.println("训练集条目数:"+trainData.count());
//        System.out.println("测试集条目数:"+testData.count());

        //2.自主选择两种回归算法并初始化参数(5分)
        //逻辑回归
        LogisticRegression lr = new LogisticRegression()
                .setFeatureCols(features)
                .setLabelCol(label)
                .setPredictionCol("pred")
                .setPredictionDetailCol("pred_detail")
                .setMaxIter(5)
                .setNumThreads(1);

        BatchOperator<?> lr_transform = lr.fit(trainData).transform(testData);

        //决策树分类
        DecisionTreeClassifier dc = new DecisionTreeClassifier()
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setLabelCol(label)
                .setFeatureCols(features)
                .setMaxBins(5)
                .setMaxDepth(5);

        BatchOperator<?> dc_transform = dc.fit(trainData).transform(testData);


        //3.使用合适的调参工具进行模型参数调优，并输出调优的中间结果(5分)
        MultiClassClassificationTuningEvaluator multiClassClassificationTuningEvaluator = new MultiClassClassificationTuningEvaluator()
                .setLabelCol(label)
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setTuningMultiClassMetric(TuningMultiClassMetric.ACCURACY);

        //网格搜索
        ParamGrid paramGrid1 = new ParamGrid()
                .addGrid(lr, LogisticRegression.L_1, new Double[] {1.0, 0.99, 0.98})
                .addGrid(lr, LogisticRegression.MAX_ITER, new Integer[] {1, 2, 3});

        GridSearchCV cv1 = new GridSearchCV()
                .setEstimator(lr)
                .setParamGrid(paramGrid1)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        PipelineModel lr_bestPipelineModel = cv1.fit(trainData).getBestPipelineModel();
        BatchOperator<?> lr_transform1 = lr_bestPipelineModel.transform(testData);

        ParamGrid paramGrid2 = new ParamGrid()
                .addGrid(dc, DecisionTreeClassifier.MAX_DEPTH, new Integer[] {1, 2, 3})
                .addGrid(dc, DecisionTreeClassifier.MAX_BINS, new Integer[] {1, 2, 3});

        GridSearchCV cv2 = new GridSearchCV()
                .setEstimator(dc)
                .setParamGrid(paramGrid2)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        PipelineModel dc_bestPipelineModel = cv2.fit(trainData).getBestPipelineModel();
        BatchOperator<?> dc_transform1 = dc_bestPipelineModel.transform(testData);
        //4.评估模型的拟合情况，至少使用两种指标(如精确率、召回率等)(10分)
        EvalBinaryClassBatchOp metrics = new EvalBinaryClassBatchOp()
                .setLabelCol(label)
                .setPredictionDetailCol("pred_detail");

        BinaryClassMetrics lr_binaryClassMetrics = metrics.linkFrom(lr_transform1).collectMetrics();
        BinaryClassMetrics dc_binaryClassMetrics = metrics.linkFrom(dc_transform1).collectMetrics();

        System.out.println("lr:精确率:"+lr_binaryClassMetrics.getAccuracy()+"召回率:"+lr_binaryClassMetrics.getMacroRecall());
        System.out.println("dc:精确率:"+dc_binaryClassMetrics.getAccuracy()+"召回率:"+dc_binaryClassMetrics.getMacroRecall());
        if(lr_binaryClassMetrics.getAccuracy()>dc_binaryClassMetrics.getAccuracy()){
            System.out.println("逻辑回归较好");
        }else{
            System.out.println("决策树分类器较好");
        }
        //5.使用Alink工具可视化模型的拟合效果，并使用较优模型预测样本并保存结果(5分)
//        lr_binaryClassMetrics.saveKSAsImage("data/yk9/lr/ks.jpg",true);
//        dc_binaryClassMetrics.saveKSAsImage("data/yk9/dc/ks.jpg",true);
//
//        lr_binaryClassMetrics.savePrecisionRecallCurveAsImage("data/yk9/lr/pre.jpg",true);
//        dc_binaryClassMetrics.savePrecisionRecallCurveAsImage("data/yk9/dc/pre.jpg",true);
//
//        lr_binaryClassMetrics.saveRocCurveAsImage("data/yk9/lr/roc.jpg",true);
//        dc_binaryClassMetrics.saveRocCurveAsImage("data/yk9/dc/roc.jpg",true);

        BatchOperator<?> transform = dc_bestPipelineModel.transform(udfBatchOp);
        dc_bestPipelineModel.save("data/yk9/dc_bestPipelineModel.csv");
        CsvSinkBatchOp csvSink = new CsvSinkBatchOp()
                .setFilePath("data/yk9/bestPipelineModel.csv")
                .setNumFiles(1)
                .setOverwriteSink(true);
        transform.link(csvSink);
        System.out.println("保存成功");
        BatchOperator.execute();



    }

    public static class SubstringFunction extends ScalarFunction {
        public int eval(int ev) {
            if(ev>=100){
                return 1;
            }
            return 0;
        }
    }
}
